# Copyright (C) 2009 Samuel Abels, http://debain.org
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

import sys, unittest, shutil, os.path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', 'src')))

def suite():
    tests = ['testStore']
    return unittest.TestSuite(map(StoreTest, tests))

from SpiffWarehouse import Store, Text, Xml, MultiBlob

class StoreTest(unittest.TestCase):
    def setUp(self):
        try:
            os.remove('testrepo')
        except:
            pass


    def tearDown(self):
        try:
            os.remove('testrepo')
        except:
            pass


    def testStore(self):
        # Create a repository.
        store = Store('testrepo')

        # Check the status.
        self.assert_(len(store.status()) == 0)

        # Add some data.
        text = 'Test Content'
        msg  = 'Added "hello".'
        store.add('hello', Text(text))
        self.assert_(len(store.status()) == 1)
        store.commit(msg)
        self.assert_(len(store.status()) == 0)
        self.assert_(store.exists('hello'))
        self.assert_(store.exists('hello', deleted = True))

        blob = store.get('hello')
        self.assert_(isinstance(blob, Text))
        self.assert_(str(blob) == text, str(blob))

        # Re-load the repository.
        store = Store('testrepo')

        # Receive the test item.
        blob = store.get('hello')
        self.assert_(isinstance(blob, Text))
        self.assert_(str(blob) == text)

        # Update some data.
        text2 = 'Hello Edited World!'
        store.add('hello', Xml(text2))
        store.add('hello2', Text('Hello Other World!'))
        self.assert_(len(store.status()) == 2)
        store.commit('Updated.')
        self.assert_(len(store.status()) == 0)

        # Check all items.
        blob = store.get('hello', revision = 1)
        self.assert_(isinstance(blob, Text))
        self.assert_(str(blob) == text)

        blob = store.get('hello', revision = 2)
        self.assert_(isinstance(blob, Xml))
        self.assert_(str(blob) == text2)

        blob = store.get('hello')
        self.assert_(isinstance(blob, Xml))
        self.assert_(str(blob) == text2)

        blob = store.get('hello2')
        self.assert_(isinstance(blob, Text))
        self.assert_(str(blob) == 'Hello Other World!')

        blob = store.get_data('hello2')
        self.assert_(str(blob) == 'Hello Other World!')

        # Test multi-part blobs.
        mb = MultiBlob(one = Text("test1\n\n\x00test1"), two = Xml('test2'))
        store.add('hello', mb)
        store.commit('Added a multiblob.')

        blob = store.get('hello')
        self.assert_(isinstance(blob, MultiBlob), type(blob))
        one = blob.get('one')
        self.assert_(isinstance(one, Text))
        self.assert_(one.bin() == "test1\n\n\x00test1")
        self.assert_(str(one) == "test1\n\ntest1")
        two = blob.get('two')
        self.assert_(isinstance(two, Xml))
        self.assert_(str(two) == 'test2')

        # Delete a blob.
        store.delete_from_name('hello')
        store.commit()
        self.assert_(not store.exists('hello'))
        self.assert_(store.exists('hello', deleted = True))
        self.assert_(len(store.find()) == 1)
        store.undelete('hello')
        store.commit()
        self.assert_(store.exists('hello'))
        self.assert_(store.exists('hello', deleted = True))

        # Same using an object.
        blob = store.get('hello')
        store.delete(blob)
        store.commit()
        self.assert_(not store.exists('hello'))
        self.assert_(store.exists('hello', deleted = True))
        self.assert_(len(store.find()) == 1)
        store.undelete('hello')
        store.commit()
        self.assert_(store.exists('hello'))
        self.assert_(store.exists('hello', deleted = True))

        # Test finding blobs.
        all = store.find()
        self.assert_(len(all) == 2)
        all = store.find('%ello%')
        self.assert_(len(all) == 2)
        all = store.find('hello')
        self.assert_(len(all) == 1)
        self.assert_(isinstance(all[0], MultiBlob))
        all = store.find('%2')
        self.assert_(len(all) == 1)
        self.assert_(isinstance(all[0], Text))

        # Purge a blob.
        store.purge('hello')
        store.commit()
        self.assert_(not store.exists('hello'))
        self.assert_(not store.exists('hello', deleted = True))
        self.assert_(len(store.find()) == 1)
        store.undelete('hello')
        self.assertRaises(Exception, store.commit)
        store.reset()

        # Test finding by content.
        mb = MultiBlob(one = Text("test1\n\n\x00test1"), two = Xml('test2'))
        store.add('hello', mb)
        store.commit('Added a multiblob.')
        all = store.find(content = '%test1%')
        self.assert_(len(all) == 1)
        self.assert_(all[0].get('one').bin() == "test1\n\n\x00test1")
        self.assert_(str(all[0].get('one')) == "test1\n\ntest1")

if __name__ == '__main__':
    unittest.TextTestRunner(verbosity = 2).run(suite())

import time
